package ai.fabu.dataplatform.service.impl;

import ai.fabu.dataplatform.Empty;
import ai.fabu.dataplatform.Status;
import ai.fabu.dataplatform.Task;
import ai.fabu.dataplatform.UpdateTaskRequest;
import ai.fabu.dataplatform.bean.bagdata.BagData;
import ai.fabu.dataplatform.bean.framedata.FrameData;
import ai.fabu.dataplatform.bean.groupdata.GroupData;
import ai.fabu.dataplatform.bean.task.GroupDataTask;
import ai.fabu.dataplatform.bean.topicdata.TopicData;
import ai.fabu.dataplatform.bean.util.IndexData;
import ai.fabu.dataplatform.bean.util.Pair;
import ai.fabu.dataplatform.constant.ResultCode;
import ai.fabu.dataplatform.help.Result;
import ai.fabu.dataplatform.service.ApiForFrontService;
import ai.fabu.dataplatform.service.ApiForImportBagService;
import ai.fabu.dataplatform.service.ApiForTaskService;
import ai.fabu.dataplatform.service.ApiService;
import ai.fabu.dataplatform.utils.TimeUtil;
import com.mongodb.MongoException;
import com.mongodb.client.result.UpdateResult;
import lombok.extern.slf4j.Slf4j;
import org.bson.types.ObjectId;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.mongodb.UncategorizedMongoDbException;
import org.springframework.data.mongodb.core.MongoOperations;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.retry.annotation.Backoff;
import org.springframework.retry.annotation.Retryable;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import static ai.fabu.dataplatform.constant.ConstantName.*;
import static ai.fabu.dataplatform.constant.ConstantValue.*;

@Service
@Slf4j
public class ApiForImportBagServiceImpl implements ApiForImportBagService {
    @Autowired
    private MongoOperations mongoOperations;
    @Autowired
    private ApiService apiService;
    @Autowired
    private ApiForTaskService apiForTaskService;


    @Value("${bag.data.save.path}")
    private String bagDataSavePath;
    @Value("${map.point.interval}")
    private int pointInterval;
    @Value("${mileage.point.interval}")
    private int mileagePointInterval;

    private final static Map<String, ScheduledFuture<?>> taskIdTable = new HashMap<>();
    private final static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);

    /**
     * 把 bin path 存到每个 frame data 里 以 topic 为单位
     * @param
     * @return
     */


    @Retryable(
            value = {MongoException.class, UncategorizedMongoDbException.class},
            maxAttempts = 50,
            backoff = @Backoff(value = 500L, multiplier = 1.5, maxDelay = 19000L)
    )
    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result<Long> saveBinData(@NotNull IndexData indexData, @NotNull List<String> binPathList) {
        Result<Long> result = new Result<>();
        String groupName = indexData.getGroupName();
        String bagName = indexData.getBagName();
        String topicName = indexData.getTopicName();
        int frameNo = indexData.getFrameNo();
        log.info("saveBinData: group name: {}, bag name: {}, topic name: {}, frame no: {}, bin path list size: {}", groupName, bagName, topicName, frameNo, binPathList.size());
        if(groupName.isEmpty() || bagName.isEmpty() || topicName.isEmpty()){
            log.error("saveBinData: group name or bag name or topic name is empty");
            result.returnWithoutValue(ResultCode.LACK_PARAM_ERROR);
            return result;
        }

        Result groupRealExistResult = isGroupRealExist(groupName);
        if(groupRealExistResult.getCode() != ResultCode.SUCCESS.getCode()){
            log.error("saveBinData: error: {}", groupRealExistResult.getMsg());
            result.returnWithOutValue(groupRealExistResult);
            return result;
        }

        Query query = new Query();
        query.addCriteria(Criteria.where(GROUP_NAME).is(groupName)).addCriteria(Criteria.where(BAG_NAME).is(bagName));
        // bag 可能已经存在了
        if(!mongoOperations.exists(query, BagData.class)){
            mongoOperations.insert(BagData.staticBagData(groupName, bagName));
        }
        // topic 一定不存在
        mongoOperations.insert(TopicData.staticTopicData(groupName, bagName, topicName));
        List<FrameData> frameDataList = new ArrayList<>();
        for(int i = 0; i < binPathList.size(); ++i){
            frameDataList.add(FrameData.staticFrameData(groupName, bagName, topicName, i+1, binPathList.get(i)));
        }
        // frame 一定不存在
        mongoOperations.insert(frameDataList, FrameData.class);
        result.returnWithValue(ResultCode.SUCCESS, ((long) binPathList.size()));
        return result;
    }


    @Retryable(
            value = {MongoException.class, UncategorizedMongoDbException.class},
            maxAttempts = 50,
            backoff = @Backoff(value = 500L, multiplier = 1.5, maxDelay = 17000L)
    )
    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result<Long> saveFramesMetadata(@NotNull IndexData indexData, @NotNull List<Pair<String, byte[]>> metadataList) {
        Result<Long> result = new Result<>();
        String groupName = indexData.getGroupName();
        String bagName = indexData.getBagName();
        String topicName = indexData.getTopicName();
        int frameNo = indexData.getFrameNo();
        log.info("saveFramesMetadata: group name: {}, bag name: {}, topic name: {}, frame no: {}, metadata size: {}", groupName, bagName, topicName, frameNo, metadataList.size());
        if(groupName.isEmpty() || bagName.isEmpty()|| topicName.isEmpty()){
            log.error("saveFramesMetadata: group name or bag name or topic name is empty");
            result.returnWithoutValue(ResultCode.LACK_PARAM_ERROR);
            return result;
        }

        Result groupRealExistResult = isGroupRealExist(groupName);
        if(groupRealExistResult.getCode() != ResultCode.SUCCESS.getCode()){
            result.returnWithOutValue(groupRealExistResult);
            return result;
        }

        long count = 0L;
        for(int i = 0; i < metadataList.size(); ++i){
            Query query = new Query();
            query.addCriteria(Criteria.where(GROUP_NAME).is(groupName)).addCriteria(Criteria.where(BAG_NAME).is(bagName)).addCriteria(Criteria.where(TOPIC_NAME).is(topicName)).addCriteria(Criteria.where(FRAME_NO).is(i+1));
            Update update = new Update();
            update.set(METADATA_POINT+metadataList.get(i).getKey(), metadataList.get(i).getValue());
            update.set(METADATA_UPDATE_TIME, TimeUtil.getNowTimestampLong().toString().getBytes());
            UpdateResult updateResult = mongoOperations.updateFirst(query, update, FrameData.class);
            count += updateResult.getModifiedCount();
        }
        result.returnWithValue(ResultCode.SUCCESS, count);
        return result;
    }


    public void addTaskScheduleTimer(String taskId, int time){
        ScheduledFuture<?> schedule = scheduledExecutorService.schedule(() -> {
            ObjectId id = new ObjectId(taskId);
            Query query = new Query(Criteria.where("_id").is(id));
            query.addCriteria(Criteria.where(COMPLETE_STATUS).is(PROCESSING_STATUS));
            Update update = new Update();
            update.set(COMPLETE_STATUS, PROCESS_FAILURE_STATUS);
            mongoOperations.updateFirst(query, update, GroupDataTask.class);
            taskIdTable.remove(taskId);
            log.warn("task: {}, has been cancelled", taskId);
        }, time, TimeUnit.MINUTES);
        taskIdTable.put(taskId, schedule);
    }

    public void addTaskScheduleTimer(String taskId){
        addTaskScheduleTimer(taskId, 30);
    }


    public boolean updateTaskScheduleTimer(String taskId, int status){
        ScheduledFuture<?> schedule = taskIdTable.get(taskId);
        if(schedule == null){
            log.error("updateTaskScheduleTimer: task id : {} not exists", taskId);
            return true;
        }
        boolean isCancel = schedule.cancel(false);
        if(status != 0)
            addTaskScheduleTimer(taskId);
        else
            taskIdTable.remove(taskId);
        return isCancel;
    }


    @Retryable(
            value = {MongoException.class, UncategorizedMongoDbException.class},
            maxAttempts = 23,
            backoff = @Backoff(value = 500L, multiplier = 1.5, maxDelay = 3000L)
    )
    @Transactional(rollbackFor = Exception.class)
    @Override
    public Result<Task> getTask(@NotBlank String source) {
        Result<Task> result = new Result<>();
        log.info("getTask: source: {}", source);
        Query query = new Query();
        query.addCriteria(Criteria.where(COMPLETE_STATUS).is(PROCESS_NOT_START_STATUS));
        query.fields().include(MAP_NAME).include(VEHICLE_NAME).include(GROUP_NAME).include(GROUP_PATH).include(METADATA).include(DONE_BAG_NUM).include(TOTAL_BAG_NUM);
        Update update = new Update();
        update.set(SOURCE, source);
        update.set(COMPLETE_STATUS, PROCESSING_STATUS);
        update.set(UPDATE_TIME, TimeUtil.getNowTimestampLong());
        GroupDataTask groupDataTask = mongoOperations.findAndModify(query, update, GroupDataTask.class);
        if(groupDataTask == null){
            result.returnWithoutValue(ResultCode.NO_DATA);
            return result;
        }
        Status status = Status.newBuilder().setDoneBagNum(groupDataTask.getDoneBagNum()).setTotalBagNum(groupDataTask.getTotalBagNum()).build();
        Task task = Task.newBuilder().setGroupName(groupDataTask.getGroupName()).setGroupPath(groupDataTask.getGroupPath()).setId(groupDataTask.getId().toHexString()).setStatus(status)
                .setMapName(groupDataTask.getMapName()).setVehicleName(groupDataTask.getVehicleName()).setSavePath(bagDataSavePath+groupDataTask.getGroupName()).build();

        // compatible fow low version
        query = new Query();
        query.addCriteria(Criteria.where(GROUP_NAME).is(groupDataTask.getGroupName()));
        if(!mongoOperations.exists(query, GroupData.class)){
            GroupData groupData = GroupData.staticGroupData(groupDataTask.getGroupName(), groupDataTask.getGroupPath(), groupDataTask.getMetadata());
            mongoOperations.insert(groupData);
        }

        result.returnWithValue(ResultCode.SUCCESS, task);
        return result;
    }


    @Override
    public Result updateTask(@NotNull UpdateTaskRequest request) {
        Result<Empty> result = new Result<>();
        Query query = new Query();
        String taskId = request.getId();
        Status status = request.getStatus();
        ObjectId objectId = new ObjectId(taskId);
        String bagName = request.getBagName();
        log.info("updateTask: task id: {}, status: {}, bag done num: {}, bag total num: {}, bag name: {}", taskId, status.getStatus(), status.getDoneBagNum(), status.getTotalBagNum(), bagName);

        query.addCriteria(Criteria.where("_id").is(objectId)).addCriteria(Criteria.where(COMPLETE_STATUS).is(PROCESSING_STATUS));
        //query.fields().include(GROUP_NAME);
        Update update = new Update();
        // successful
        if(status.getStatus().getNumber() == 0){
//            update.set(COMPLETE_STATUS, PROCESSED_STATUS);
            update.set(CURRENT_BAG_NAME, STRING_DEFAULT);
            if(status.getTotalBagNum() == status.getDoneBagNum())
                update.set(COMPLETE_STATUS, PROCESSED_STATUS);
        }
        else if(status.getStatus().getNumber() == 2){
            update.set(COMPLETE_STATUS, PROCESS_FAILURE_STATUS);
            update.set(CURRENT_BAG_NAME, bagName);
        }
        else
            update.set(CURRENT_BAG_NAME, bagName);
        update.set(DONE_BAG_NUM, status.getDoneBagNum());
        update.set(TOTAL_BAG_NUM, status.getTotalBagNum());
        update.set(UPDATE_TIME, TimeUtil.getNowTimestampLong());
        UpdateResult updateResult = mongoOperations.updateFirst(query, update, GroupDataTask.class);
        //GroupDataTask groupDataTask = mongoOperations.findAndModify(query, update, GroupDataTask.class);
        if(updateResult.getMatchedCount() == 0){
            log.error("updateTask: update task error, (no such task: {}, or task is cancelled)", taskId);
            result.returnWithoutValue(ResultCode.PARAM_ERROR);
            return result;
        }
        result.returnWithoutValue(ResultCode.SUCCESS);
        return result;
    }

    public Result isGroupRealExist(@NotBlank String groupName) {
        Result result = new Result();
        if(groupName.isEmpty()){
            result.returnWithoutValue(ResultCode.SUCCESS);
            return result;
        }

        Query query = new Query(Criteria.where(GROUP_NAME).is(groupName));
        query.fields().include(COMPLETE_STATUS);
        GroupDataTask groupDataTask = mongoOperations.findOne(query, GroupDataTask.class);
        if(groupDataTask == null){
            log.error("isGroupRealExist: param error, (no such group)");
            result.returnWithoutValue(ResultCode.PARAM_ERROR);
            return result;
        }
        if(groupDataTask.getCompleteStatus() == 0){
            log.error("isGroupRealExist: group: {} never start yet", groupName);
            result.returnWithoutValue(ResultCode.TASK_NOT_START_ERROR);
            return result;
        }
        if(groupDataTask.getCompleteStatus() == 2){
            log.error("isGroupRealExist: group: {} has already been completed", groupName);
            result.returnWithoutValue(ResultCode.TASK_COMPLETE_ERROR);
            return result;
        }
        if(groupDataTask.getCompleteStatus() == 3){
            log.error("isGroupRealExist: group: {} has already been cancel", groupName);
            result.returnWithoutValue(ResultCode.TASK_FAILURE_ERROR);
            return result;
        }
        result.returnWithoutValue(ResultCode.SUCCESS);
        return result;
    }



//    @Override
//    public Result generateGroupMapAndBagRecord(@NotBlank String taskId) {
//        log.info("generateGroupMapAndBagRecord: task id: {}", taskId);
//        Query query = new Query();
//        ObjectId id = new ObjectId(taskId);
//        query.addCriteria(Criteria.where("_id").is(id));
//        query.fields().include(GROUP_NAME);
//        GroupDataTask groupDataTask = mongoOperations.findOne(query, GroupDataTask.class);
//        if(groupDataTask == null){
//            log.error("generateGroupMapAndBagRecord: group data task error");
//            throw new RuntimeException("generateGroupMapAndBagRecord: mongo db error");
//        }
//        String groupName = groupDataTask.getGroupName();
//        Result result;
//        try{
//            Result mapResult = apiForFrontService.generateGroupMapByName(groupName);
//            if(mapResult.getCode() == ResultCode.SUCCESS.getCode()){
//                log.info("generateGroupMapAndBagRecord: generate static map success");
//            }
//            else{
//                result = mapResult;
//                log.error("generateGroupMapAndBagRecord: generate static map fail. msg: {}", mapResult.getMsg());
//            }
//        } catch (Exception e){
//            log.error("generateGroupMapAndBagRecord: generate static map fail");
//            e.printStackTrace();
//        }
//        Result recordResult = apiForFrontService.generateBagRecordByName(groupName);
//        result = recordResult;
//        if(recordResult.getCode() == ResultCode.SUCCESS.getCode()){
//            log.info("generateGroupMapAndBagRecord: generate bag record success");
//        }
//        else{
//            log.error("generateGroupMapAndBagRecord: generate bag record fail. msg: {}", recordResult.getMsg());
//        }
//        return result;
//    }

    private String getGroupNameByTaskId(@NotBlank String taskId){
        Query query = new Query();
        ObjectId id = new ObjectId(taskId);
        query.addCriteria(Criteria.where("_id").is(id));
        query.fields().include(GROUP_NAME);
        GroupDataTask groupDataTask = mongoOperations.findOne(query, GroupDataTask.class);
        if(groupDataTask == null){
            log.error("getGroupNameByTaskId: group data task error");
            throw new RuntimeException("getGroupNameByTaskId: mongo db error");
        }
        return groupDataTask.getGroupName();
    }

    @Override
    public Result generateGroupMapAndMileage(@NotBlank String taskId) {
        String groupName = getGroupNameByTaskId(taskId);
        String methodName = new Exception().getStackTrace()[0].getMethodName();
        log.info("{}: group name: {}", methodName, groupName);
        Result result = apiForTaskService.generateGroupMapAndMileageByName(groupName);
        return result;
    }

    @Override
    public Result generateBagRecord(@NotBlank String taskId) {
        String groupName = getGroupNameByTaskId(taskId);
        String methodName = new Exception().getStackTrace()[0].getMethodName();
        log.info("{}: group name: {}", methodName, groupName);
        Result result = apiForTaskService.generateBagRecordByName(groupName);
        return result;
    }

    @Override
    public Result generateBagGPS(@NotBlank String taskId) {
        String groupName = getGroupNameByTaskId(taskId);
        String methodName = new Exception().getStackTrace()[0].getMethodName();
        log.info("{}: group name: {}", methodName, groupName);
        Result result = apiForTaskService.generateBagGpsByName(groupName);
        return result;
    }

    @Retryable(
            value = {MongoException.class, UncategorizedMongoDbException.class},
            maxAttempts = 30,
            backoff = @Backoff(delay = 100L, maxDelay = 1000L, random = true)
    )
    @Override
    public Result updateGroupTaskCompleteStatus(@NotBlank String taskId, @NotNull Byte completeStatus) {
        String methodName = new Exception().getStackTrace()[0].getMethodName();
        log.info("{}: task id: {}, complete status: {}", methodName, taskId, completeStatus);
        Result result = new Result();
        ObjectId objectId = new ObjectId(taskId);
        Query query = new Query();
        query.addCriteria(Criteria.where("_id").is(objectId));
        Update update = new Update();
        update.set(COMPLETE_STATUS, completeStatus);
        update.set(UPDATE_TIME, TimeUtil.getNowTimestampLong());
        UpdateResult updateResult = mongoOperations.updateFirst(query, update, GroupDataTask.class);
        if(updateResult.getModifiedCount() == 1){
            result.returnWithoutValue(ResultCode.SUCCESS);
        }
        else{
            log.error("{}: update group task error", methodName);
            result.returnWithoutValue(ResultCode.SYSTEM_ERROR);
        }
        return result;
    }
}
